package com.study.flink.datastream

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
  * TimeWindow
  *
  * @author: stephen.shen
  * @create: 2019-05-27 17:47
  */
object FlinkTimeWindowDemo {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val socketDStream: DataStream[String] = env.socketTextStream("localhost",1234)
    // 导入隐式转换
    import org.apache.flink.api.scala._
    val mapDStream: DataStream[(String, Int)] = socketDStream.map(e => {
      val strings: Array[String] = e.split(" ")
      (strings(0), strings(1).toInt)
    })
    val keyDStream: KeyedStream[(String, Int), Tuple] = mapDStream.keyBy(0)
    // 滚动窗口 每 3 秒对进入该窗口的所有相同key 的数据进行reduce 和 print 操作
    val windowDStream: WindowedStream[(String, Int), Tuple, TimeWindow] = keyDStream.timeWindow(Time.seconds(3))
    // 滑动窗口 每 2 秒对进入该窗口的所有数据进行前 4 秒数据的 reduce 和 print 操作
    //val windowDStream: WindowedStream[(String, Int), Tuple, TimeWindow] = keyDStream.timeWindow(Time.seconds(4),Time.seconds(2))

    val reduceDStream: DataStream[(String, Int)] = windowDStream.reduce((e1,e2)=>(e1._1,e1._2+e2._2))
    reduceDStream.print()


    env.execute("Flink TimeWindow Demo")
  }
}
